package cn.karent.core;

import cn.karent.common.JsonUtils;
import cn.karent.common.Result;
import cn.karent.common.RocketMQConfig;
import cn.karent.processor.ConfigurableProcessor;
import cn.karent.processor.Processor;
import io.micrometer.tracing.Span;
import io.micrometer.tracing.Tracer;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.springframework.context.ApplicationContext;
import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author wanshengdao
 * @date 2025/3/20
 */
@Slf4j
@RestController
@RequiredArgsConstructor
public class ConfigController {

    public static final Consumer DEFAULT_CONSUMER = new Consumer();
    private final Map<String, Consumer> consumers = new ConcurrentHashMap<>();

    private final ApplicationContext applicationContext;

    private final RocketMQConfig rocketMQConfig;
    
    private final Tracer tracer;

    private final ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(2, new ThreadFactory() {

        private final AtomicInteger cnt = new AtomicInteger(0);
        @Override
        public Thread newThread(@NotNull Runnable r) {
            return new Thread(r, "scheduled-task-" + cnt.getAndIncrement());
        }
    });

    /**
     * RocketMQ监听配置
     *
     * @param cmd 请求
     * @return 配置结果
     */
    @PostMapping("/listen")
    public Result<String> listen(@Valid @RequestBody ConfigCmd cmd) throws MQClientException {
        Assert.isTrue(!consumers.containsKey(cmd.getTopic()), String.format("topic [%s] 已经监听, 请先关闭!", cmd.getTopic()));
        Consumer consumer = Consumer.builder()
                .endpoints(rocketMQConfig.getEndpoints())
                .group(cmd.getGroup())
                .topic(cmd.getTopic())
                .processor(getProcessor(cmd))
                .applicationContext(applicationContext)
                .build();
        consumer.listen();
        consumers.put(cmd.getTopic(), consumer);
        scheduledShutdownIfNeed(cmd, consumer);
        return Result.ok();
    }

    @Nullable
    private Processor getProcessor(ConfigCmd cmd) {
        ConfigCmd.ProcessorConfig config = cmd.getProcessor();
        if (config == null) {
            return null;
        }
        String beanName = config.getName();
        Processor processor = applicationContext.getBean(beanName, Processor.class);
        if (processor instanceof ConfigurableProcessor<?> configurable) {
            Map<String, Object> map = config.getConfig();
            String str = Optional.ofNullable(map)
                    .map(JsonUtils::toString)
                    .orElse(null);
            configurable.config(str);
        }
        return processor;
    }

    private void scheduledShutdownIfNeed(ConfigCmd cmd, Consumer consumer) {
        if (cmd.getDuration() != null) {
            log.info("该topic的监听将会在 {}ms后释放", cmd.getDuration().toMillis());
            scheduled.schedule(() -> {
                Span newSpan = tracer.nextSpan().name("scheduled-task").start();
                try (Tracer.SpanInScope ws = tracer.withSpan(newSpan)) {
                    log.info("定时任务触发执行");
                    consumer.shutdown();
                    consumers.remove(cmd.getTopic());
                } catch (Exception e) {
                    log.error("定时任务调度异常: ", e);
                } finally {
                    newSpan.end();
                }
            }, cmd.getDuration().toMillis(), TimeUnit.MILLISECONDS);
        }
    }

    /**
     * RocketMQ停止监听topic
     *
     * @return 关闭结果
     */
    @PostMapping("/stop")
    public Result<String> stop(@RequestParam("topic") String topic) {
        consumers.getOrDefault(topic, DEFAULT_CONSUMER).shutdown();
        consumers.remove(topic);
        return Result.ok();
    }

}
